package com.leyantech.ecp.agenx.session.client;

import com.leyantech.utility.Logger;
import com.leyantech.utility.guava.cache.GuavaCacheUtils;
import com.leyantech.utility.statsd.StatsDClientFactory;

import com.google.common.base.Preconditions;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import com.google.common.collect.ImmutableSet;
import com.google.inject.Inject;
import com.google.inject.Singleton;
import com.google.inject.name.Named;
import com.timgroup.statsd.StatsDClient;
import io.opentracing.References;
import io.opentracing.Scope;
import io.opentracing.Span;
import io.opentracing.SpanContext;
import io.opentracing.Tracer;
import io.opentracing.contrib.concurrent.TracedExecutorService;
import io.opentracing.contrib.kafka.TracingKafkaUtils;
import io.opentracing.tag.Tags;
import io.opentracing.util.GlobalTracer;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;

import java.util.Collection;
import java.util.Collections;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.function.Predicate;
import java.util.function.Supplier;
import java.util.stream.Collectors;

/**
 * @author eahau, {@literal <eahau@foxmail.com>}
 * @date 2022-11-01.
 */
@Singleton
public class SessionClient {

  private final Logger logger = new Logger(SessionClient.class);

  private final ExecutorService executorService;

  private final Consumer<String, String> consumer;

  private final Supplier<LoadingCache<String, Set<Assistant>>> assistantCache;

  private Map<String, java.util.function.Consumer<Assistant>> actionListeners = Collections.emptyMap();

  @Inject(optional = true)
  private void setActionListeners(Map<String, java.util.function.Consumer<Assistant>> listeners) {
    if (listeners != null) {
      this.actionListeners = listeners
          .entrySet()
          .stream()
          .filter(entry -> entry.getKey() != null && entry.getValue() != null)
          .collect(Collectors.toMap(Entry::getKey,
              entry -> event -> this.executorService.execute(() -> {
                final String action = event.getAction();
                final String key = String.join(".", "assistantListener", action);
                try {
                  final long start = System.nanoTime();
                  entry.getValue().accept(event);
                  client.time(key, System.nanoTime() - start);
                  client.increment(key + ".succeed");
                } catch (Exception e) {
                  client.increment(key + ".error");
                  logger.error("call listener error", e);
                }
              }))
          );
    }
  }


  private volatile boolean running = true;

  static final StatsDClient client = StatsDClientFactory.getClient();

  @Inject
  public SessionClient(@Named("session-client") ExecutorService executorService,
      @Named("session-client") Consumer<String, String> consumer,
      @Named("session-client") CacheLoader<String, Set<Assistant>> cacheLoader) {
    this.executorService = executorService instanceof TracedExecutorService
        ? executorService : new TracedExecutorService(executorService, GlobalTracer.get());
    this.consumer = consumer;
    this.assistantCache = GuavaCacheUtils
        .buildDynamicLoadingCacheSupplier("session-client", cacheLoader);

    start();
  }

  /**
   * 店铺在线客服.
   */
  public ImmutableSet<Assistant> listAssistant(String storeId) {
    Preconditions.checkArgument(StringUtils.isNotBlank(storeId), "storeId is blank");

    return ImmutableSet.copyOf(assistantCache.get().getUnchecked(storeId));
  }

  /**
   * 店铺在线客服.
   */
  public Optional<Assistant> getOnlineAssistant(String storeId, String assistantId) {
    Preconditions.checkArgument(StringUtils.isNotBlank(storeId), "storeId is blank");
    Preconditions.checkArgument(StringUtils.isNotBlank(assistantId), "assistantId is blank");
    return listAssistant(storeId)
        .stream()
        .filter(assistant -> assistantId.equals(assistant.getAssistantId()))
        .findFirst();
  }

  /**
   * 客服是否在线.
   */
  public boolean isAssistantOnline(String storeId, String assistantId) {
    return getOnlineAssistant(storeId, assistantId).isPresent();
  }

  void start() {
    new Thread(() -> {
      while (running) {
        try {
          consumer.poll(1000L).forEach(record ->
              executorService.execute(() -> {
                try {
                  onAssistantEventChanged(record);
                } catch (Exception e) {
                  logger.error("onAssistantEventChanged failed", e);
                }
              })
          );
        } catch (Exception e) {
          logger.error("poll assistant-event failed", e);
        }
      }
    }, "assistant-event-poller")
        .start();
  }

  void stop() {
    running = false;
    try {
      executorService.shutdown();
      executorService.awaitTermination(30L, TimeUnit.SECONDS);
    } catch (Exception ignore) {
    }
    consumer.close();
  }

  void onAssistantEventChanged(ConsumerRecord<String, String> record) {

    if (StringUtils.contains(record.key(), "warmup")) {
      return;
    }

    final Assistant assistant = Assistant.fromJson(record.value());
    if (StringUtils.equalsAny("warmup", assistant.getStoreId(), assistant.getAssistantId())) {
      return;
    }

    final Tracer tracer = GlobalTracer.get();
    final SpanContext parentContext = TracingKafkaUtils
        .extractSpanContext(record.headers(), tracer);

    final Tracer.SpanBuilder spanBuilder = tracer
        .buildSpan("consume-assistant-event")
        .withTag(Tags.SPAN_KIND.getKey(), Tags.SPAN_KIND_CONSUMER);

    if (parentContext != null) {
      spanBuilder.addReference(References.FOLLOWS_FROM, parentContext);
    }

    final String action = assistant.getAction();
    final String key = String.join(".", "assistant", action);

    final Span span = spanBuilder.start();
    try (final Scope ignore = tracer.scopeManager().activate(span)) {
      final long start = System.nanoTime();
      onAssistantEventChanged(assistant);
      client.time(key, System.nanoTime() - start);
      client.increment(key + ".succeed");
    } catch (Exception e) {
      client.increment(key + ".error");
      logger.error("onAssistantEventChanged failed", e);
    } finally {
      span.finish();
    }
  }

  private void callListener(Assistant event) {
    final String action = event.getAction();
    final java.util.function.Consumer<Assistant> assistantConsumer = actionListeners.get(action);
    if (assistantConsumer != null) {
      assistantConsumer.accept(event);
    }
  }

  private void onShutdown(Assistant shutdownEvent) {
    final String nodeId = shutdownEvent.getNodeId();
    Preconditions.checkArgument(StringUtils.isNotBlank(nodeId), "nodeId is blank.");

    final String action = shutdownEvent.getAction();
    final java.util.function.Consumer<Assistant> assistantConsumer = actionListeners.get(action);

    assistantCache.get()
        .asMap()
        .values()
        .stream()
        .flatMap(Collection::stream)
        .filter(assistant -> nodeId.equals(assistant.getNodeId()))
        .forEach(assistant -> {
          final String storeId = assistant.getStoreId();
          assistantCache.get()
              .getUnchecked(storeId)
              .remove(assistant);
          logger.debug("onShutdown remove sid {} aid {}", storeId, assistant.getAssistantId());
          if (assistantConsumer != null) {
            assistant.setAction(action);
            assistant.setTimestamp(shutdownEvent.getTimestamp());
            assistantConsumer.accept(assistant);
          }
        });
  }

  public void onAssistantEventChanged(Assistant newAssistant) {

    logger.info("received assistant event {}", newAssistant);

    if (newAssistant.isShutdown()) {
      onShutdown(newAssistant);
      return;
    }

    final String storeId = newAssistant.getStoreId();
    final String assistantId = newAssistant.getAssistantId();

    Preconditions.checkArgument(StringUtils.isNotBlank(storeId), "storeId is blank.");
    Preconditions.checkArgument(StringUtils.isNotBlank(assistantId), "assistantId is blank.");

    final Set<Assistant> assistantSet = assistantCache.get().getUnchecked(storeId);

    final Optional<Assistant> oldAssistant = assistantSet.stream()
        .filter(newAssistant::equals)
        .findFirst();

    oldAssistant.ifPresent(assistant -> logger.info("lastAssistant {}", assistant));

    final Predicate<Assistant> isNewEvent = assistant ->
        newAssistant.getTimestamp() > assistant.getTimestamp();

    if (newAssistant.isLogout()) {
      // 客服登出，需要根据节点 id 删除
      final String nodeId = newAssistant.getNodeId();
      if (StringUtils.isBlank(nodeId)) {
        logger.warn("nodeId is blank");
        return;
      }
      oldAssistant.filter(isNewEvent.and(assistant -> nodeId.equals(assistant.getNodeId())))
          .ifPresent(assistant -> {
            if (assistantSet.remove(assistant)) {
              logger.info("{} remove logout assistant {} succeed", storeId, assistantId);
              callListener(newAssistant);
            }
          });
    } else {
      // 比较时间戳或缓存不存在
      final boolean needUpdate = oldAssistant.map(isNewEvent::test).orElse(true);
      if (needUpdate) {
        oldAssistant.ifPresent(assistantSet::remove);
        assistantSet.add(newAssistant);
        logger.info("{} update {} assistant {} succeed, last {}",
            storeId, newAssistant.getAction(), assistantId, oldAssistant.orElse(null));
        callListener(newAssistant);
      }
    }
  }

}
